package base.test8.mapjoin;


import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;

public class MapJoinMapper extends Mapper<LongWritable,Text, Text, NullWritable> {

    private HashMap<String,String> pdMap = new HashMap<>();

    Text outK = new Text();

    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取缓存文件，并把文件内容封装到集合中

        //获取缓存文件URI路径
        URI[] cacheFiles = context.getCacheFiles();

        //获取路径下的文件
        FileSystem fs = FileSystem.get(context.getConfiguration());
        FSDataInputStream fis = fs.open(new Path(cacheFiles[0]));

        //从流中读取数据
        BufferedReader reader = new BufferedReader(new InputStreamReader(fis, StandardCharsets.UTF_8));

        String line;
        //用工具类判断行是否为空，为空则结束循环
        while(StringUtils.isNotEmpty(line = reader.readLine())){

            //切割
            String[] fields = line.split("\t");

            //赋值
            pdMap.put(fields[0],fields[1]);

        }
        //关流
        IOUtils.closeStream(reader);

    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //处理order.txt
        String line = value.toString();
        String[] fields = line.split("\t");

        //获取id
        String pname = pdMap.get(fields[1]);

        //获取订单id 和订单数量
        //封装

        outK.set(fields[0] + "\t" + pname + "\t" +  fields[2]);

        context.write(outK, NullWritable.get());

    }
}
